package cn.iotab.iotkit.transport.api.access.handler.impl;

import cn.iotab.iotkit.transport.api.access.TransportAccessService;
import cn.iotab.iotkit.transport.api.access.handler.DeviceMessageSender;
import cn.iotab.iotkit.transport.api.access.handler.session.Session;
import cn.iotab.iotkit.transport.api.access.handler.support.ToTransportMessageSender;
import cn.iotab.iotkit.transport.api.access.model.message.DeviceMessage;
import cn.iotab.iotkit.transport.api.access.model.message.DeviceMessageReply;
import cn.iotab.iotkit.transport.api.access.model.message.OriginMessage;
import cn.iotab.iotkit.transport.api.access.model.message.OriginMessageReply;
import cn.iotab.iotkit.transport.api.common.id. DeviceMessageKey;
import cn.iotab.iotkit.transport.api.common.message.msg.CommonPayload;
import cn.iotab.iotkit.transport.api.common.queue.SendResultCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class DefaultDeviceMessageSender implements DeviceMessageSender {

    private final DefaultDeviceSessionManager sessionManager;
    private final DefaultDeviceReplyListener replyListener;

    private final ToTransportMessageSender sender;

    public DefaultDeviceMessageSender(TransportAccessService accessService,
                                      DefaultDeviceSessionManager sessionManager,
                                      DefaultDeviceReplyListener replyListener) {
        this.sessionManager = sessionManager;
        this.replyListener = replyListener;
        this.sender = new ToTransportMessageSender(accessService);
    }


    @Override
    public ListenableFuture<DeviceMessageReply> sendMessage(DeviceMessage message) {
        Session sessionInfo = sessionManager.getSessionByDevice(message.getProductKey(), message.getDeviceId());
        if (sessionInfo == null) return Futures.immediateFailedFuture(new IllegalStateException("device not online"));
        //
        OriginMessage originMessage = message.toOriginMessage();
        String payload = originMessage.toJson().toJSONString();
        //
        CommonPayload request = CommonPayload.of(message.getProductKey(), message.getDeviceId(), payload);

        ListenableFuture<DeviceMessageReply> reply = null;
        if (message.needAck()) {
            reply = replyListener.getReply(message);
        }

        sender.sendToDeviceRequest(sessionInfo.getSessionId(), request, new SendResultCallback<Void>() {
            @Override
            public void onSuccess(Void msg) {
                log.trace("message {} send success", payload);
            }

            @Override
            public void onError(Throwable e) {
                DeviceMessageKey key = DeviceMessageKey.of(message);
                replyListener.onException(key, e);
            }
        });
        //get reply
        return reply;
    }

    @Override
    public void sendReply(String sessionId, DeviceMessageReply reply) {

        OriginMessageReply originMessage = reply.toOriginMessageReply();
        String payload = originMessage.toJson().toJSONString();

        CommonPayload response = CommonPayload.of(reply.getProductKey(), reply.getDeviceId(), payload);
        sender.sendToDeviceResponse(sessionId, response, new SendResultCallback<Void>() {
            @Override
            public void onSuccess(Void msg) {
                log.trace("message {} send success", payload);
            }

            @Override
            public void onError(Throwable e) {
                log.warn("send deviceMessage reply failed, payload: {}", payload, e);
            }
        });
    }
}
